package com.edu.whty.common.tools.kafka;

//import com.alibaba.fastjson.JSONObject;
//import com.edu.whty.service.CourseService;
//import com.edu.whty.service.UserTextBookService;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.support.Acknowledgment;
//import org.springframework.stereotype.Component;

//@Component
public class KafkaConsumer {

//    private Logger logger = LoggerFactory.getLogger(getClass());
//
//    @Autowired
//    private UserTextBookService userTextBookService;
//
//    @Autowired
//    private CourseService courseService;
//
//    @KafkaListener(topics = "#{'${rrtbk.kafka.topics.read}'.split(',')}")
//    public void receive(ConsumerRecord record, Acknowledgment acknowledgment) {
//        logger.info("Kafka同步消息：topic-{{}},partition-{{}},offset-{{}},value-{{}}", record.topic(), record.partition(),record.offset(), record.value());
//        try {
//            CommonMessage message = (CommonMessage) JSONObject.parseObject(record.value().toString
//                    (), CommonMessage.class);
//            String table = message.getTable();
//            if ("tb_user_textbook".equals(table)) {
//                userTextBookService.syncUserTextBookBatch(message.getData(), message.getType());
//            } else if ("tb_user_textbook_chapter".equals(table)) {
//                userTextBookService.syncUserTextBookChapterBatch(message.getData(), message.getType());
//            } else if ("tb_course_res".equals(table)) {
//                courseService.syncCourseResBatch(message.getData(), message.getType());
//            } else if ("tb_course_user".equals(table)) {
//                courseService.syncCourseUserBatch(message.getData(), message.getType());
//            }
//            acknowledgment.acknowledge();
//            logger.info("同步成功。");
//        } catch (Exception e) {
//            logger.error("同步失败：" + e.getMessage());
//        }
//    }
}
